<?php
/*
 * Copyright (c) 2008 Sergey Bondari, bondari@bondari.net
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * - Redistributions of source code must retain the above copyright
 *   notice, this list of conditions and the following disclaimer.
 * - Redistributions in binary form must reproduce the above copyright
 *   notice, this list of conditions and the following disclaimer in the
 *   documentation and/or other materials provided with the distribution.
 * - The name of the author may not be used to endorse or promote products
 *   derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

namespace Mosaic\System;


/**
 * Requests processing multiplexor
 * 
 * Request execution multiplexor is the most advanced subsystem that allows parallel processing of web document 
 * requests made to the MOSAIC site in the global scope. When component executes a synchronous request
 * it is blocked until the response returns. In this situation the request multiplexor starts the execution 
 * of another request pending in the request broker queue. 
 * When the response returns, the multiplexor continues execution of the original component code. 
 * This approach reduces the amount of time a worker wasting by waiting for a response from other nodes to the minimum.
 *
 * @package System
 * @subpackage Node
 */
class RequestMultiplexor {

	/**
	 * Queue of requests to send out
	 * @var SplPriorityQueue
	 */
	private $queue;


	/**
	 * Pool of active requests that were not resolved
	 * @var array
	 */
	private $pool = array();



	/**
	 * Class constructor
	 */
	function __construct() {
		$this->queue = new \SplPriorityQueue();
		$this->queue->setExtractFlags(\SplPriorityQueue::EXTR_DATA);
	}




	/**
	 * Enqueues request/response for delivery
	 *
	 * @param ifRequestEntity Request or response to send
	 * @param mixed Callback to be called when response arrives. Parameter makes sense only for requests.
	 * @return void
	 */
	private function enqueue(ifRequestEntity $r, $callback = null) {
		if ($r instanceof Request) {
			trigger_error('Adding request ['.$r->getRequestToken().'] to the pool and queue.');
			/*
			 * Add request to pool if response will be used
			 */
			if ($r->getExecType() != ASYNC || isset($callback)) {
				$this->pool[$r->getRequestToken()] = array($r, null, $callback);
			}
	
			/*
			 * sync requests have higher priorities than async requests
			 */
			$this->queue->insert($r, $r->getExecType() == SYNC ? 2 : 3);
		} else if ($r instanceof Response) {
			trigger_error('Adding response ['.$r->getRequestToken().'] to the queue.');
			/*
			 * responses have higher priorities than requests
			 */
			$this->queue->insert($r, 1);
			
		}
	}


	/**
	 * Removes requests/responses from pool
	 *
	 * @param ifRequestEntity Request or response
	 * @return void
	 */
	private function dequeue(ifRequestEntity $r) {
		unset($this->pool[$r->getRequestToken()]);
	}




	/**
	 * Request execution multiplexor. 
	 *
	 * Function is an infinite loop. Returns when response for the given token is recieved or available.
	 *
	 * @param string Request token
	 * @return Response
	 */
	private function multiplexor($returnUponToken = NULL) {
		global $sys;

		trigger_error('Entering multiplexor');

		while (1) {
				/*
				 * Every cycle multiplexor verifies if current stack response is not yet available
				 */
				if (isset($returnUponToken) && $this->pool[$returnUponToken][1] instanceof Response) {
					trigger_error('This is an active stack response. Returning value from multiplexor.');
					return $this->pool[$returnUponToken][1];
				}

						

				/*
				 * Receive requests on these sockets
				 */
				$socketsToRead = array($sys->rbSocket);

				/*
				 * Send requests to these sockets
				 */
				$socketsToWrite = $this->queue->count() ? array($sys->rbSocket) : NULL;
				
				/*
				 * Wait for socket event
				 */
				$stat = stream_select($socketsToRead, $socketsToWrite, $except = NULL, NULL);

				/* Unexpected behaviour */
				if ($stat === false) {
					trigger_error('stream_select() did not return any sockets.',E_USER_ERROR);
					continue;
				}


				/*
				 * Process incoming traffic
				 */
				if ($socketsToRead) foreach ($socketsToRead as $socket) {
					trigger_error('Socket is ready to send data');
					if (feof($socket)) {
						throw new RuntimeException('Node intercommunication socket was closed');
					}

					$mcgi = new \Mosaic\MCGI\Packet();
					try {
						/*
						 * New request/response
						 */
						$mcgi->receive($socket);
						if ($mcgi->getHeader('RESPONSE')) {
							$response = Response::createFromMCGIPacket($mcgi);
							$this->processIncomingResponse($response);
						} else {
							$request = Request::createFromMCGIPacket($mcgi);
							$this->processIncomingRequest($request);
	
						}
					} catch (IOException $e) {
						trigger_error('IO Exception');
					}
				
				}


				/*
				 * Process outgoing traffic
				 */
				if ($socketsToWrite) foreach ($socketsToWrite as $socket) {
					trigger_error('Socket is ready to receive data');
					trigger_error('Extracting request');
					$r = $this->queue->extract();
					trigger_error('Preparing MCGI packet');
					$mcgi = $r->createMCGIPacket();
					trigger_error('Sending MCGI packet');
					$mcgi->send($socket);
				}

		}
				
		
	}


	/**
	 * Entry point from node initialization code
	 * @return void
	 */
	public function processRequests() {
		trigger_error('Enterinm multiplexing loop. Execution stack bottom.');
		$this->multiplexor();
		trigger_error('Exited multiplexor. Execution stack is empty.');
	}



	/**
	 * Process incoming request
	 *
	 * Method processes request received by multiplexor
	 *
	 * @return void
	 */
	private function processIncomingRequest(Request $req) {
		global $sys;
		trigger_error('Processing request of type '. $req->getRequestType());
		$response = new Response($req);
		switch ($req->getRequestType()) {
			/*
			 * Remote object instantiate. Accepted by object broker.
			 */
			case RequestType::REMOTE_OBJECT_INSTANTIATE:
				if ($sys->role == OBJECT_BROKER) {
					try {
						$response->setData($sys->objectBroker->instantiate($req));
					} catch (RemoteExceptionThrowable $e) {
						$response->setData($e);
					}
					$this->enqueue($response);
				} else {
					/*
					 * Should not be reached
					 */
					throw new RuntimeExeption('Wrong recipient!');
				}

			break;

			/*
			 * Remote object method call. Accepted by object broker.
			 */
			case RequestType::REMOTE_OBJECT_CALL:
				if ($sys->role == OBJECT_BROKER) {
					try {
						$response->setData($sys->objectBroker->call($req));
					} catch (RemoteExceptionThrowable $e) {
						$response->setData($e);
					}
					$this->enqueue($response);
				} else {
					/*
					 * Should not be reached
					 */
					throw new RuntimeExeption('Wrong recipient!');
				}
			break;

			/*
			 * Remote object stub code
			 */
			case RequestType::REMOTE_OBJECT_STUB:
				if ($sys->role == OBJECT_BROKER) {
					try {
						$response->setData($sys->objectBroker->getStubCode($req));
					} catch (RemoteExceptionThrowable $e) {
						$response->setData($e);
					}
					$this->enqueue($response);
				} else {
					/*
					 * Should not be reached
					 */
					throw new RuntimeExeption('Wrong recipient!');
				}
			break;


			/*
			 * Request for content. Accepted by worker.
			 */
			case RequestType::HTTP_CONTENT:
				if ($sys->role == WORKER) {
					/*
					 * Switch context
					 */
					$req->setContextTag($sys->switchContext($req->getContextTag()));

					/*
					 * Prepare content getter event
					 */
					$e = new evtContent();
					$e->setRequestToken($req->getRequestToken());
					$e->setRequestHeaders($req->getHeaders());
					$e->setRequestBody($req->getBody());
					$e->pushCallback(array($this, 'onContentReady'));

					/*
					 * Add request to pool
					 */
					$this->pool[$req->getRequestToken()] = array($req, NULL);

					/*
					 * Execute event asynchroniously
					 */
					$sys->icc->fireEventAsync($e);

				} else {
					throw new RuntimeExeption('Wrong recipient!');
				}
			break;


			/*
			 * New node up
			 */
			case RequestType::NODE_UP_MANIFEST:
				trigger_error('Loading remote manifest from "'.$req->getNodename().'"', E_USER_NOTICE);
				$sys->componentBroker->loadRemoteManifest($req->getNodeName(), $req->getManifest());
			break;

			/*
			 * Node down
			 */
			case RequestType::NODE_DOWN_MANIFEST:
				trigger_error('Recieved a node down notification from "'.$req->getNodename().'"', E_USER_NOTICE);
				$sys->componentBroker->unloadRemoteManifest($req->getNodeName());
			break;


			case RequestType::RESOURCE:
				/*
				 * Switch context
				 */
				$req->setContextTag($sys->switchContext($req->getContextTag()));

				try {
					$res = $req->getResource();
					/*
					 * Add request to pool
					 */
					$this->pool[$req->getRequestToken()] = array($req, NULL);
					$_this=$this;
					$res->pushCallback(function() use ($res, $req, $_this) { $_this->onResourceReady($res,$req); });
					$sys->icc->processResource($res);
				} catch (RemoteExceptionThrowable $e) {
					$response->setData($e);
					$this->enqueue($response);
				}
			break;

			case RequestType::RELEASE_CONTEXT:
				//$sys->switchContext($req->getContextTag());
				$sys->releaseContext($req->getContextTag());
			break;

			default:
				throw new RuntimeException('Invalid request type');
		}

	}


	/**
	 * Process incoming response
	 *
	 * Processing of response packets recieved by multiplexor
	 * @return void
	 */
	private function processIncomingResponse(Response $res) {
		global $sys;


		if (!isset($this->pool[$res->getRequestToken()])) {
			trigger_error('Response '.$res->getRequestToken().' is not of my interest');
			return;
		}

		trigger_error('Adding response to pool.');

		/*
		 * Response is placed into the pool of requests and responses.
		 */
		$this->pool[$res->getRequestToken()][1] = $res;

		/*
		 * This code processes asynchronous responses with callback.
		 */
		if (isset($this->pool[$res->getRequestToken()][2])) {
			/*
			 * Get context tag from original request
			 */
			$ctx = $this->pool[$res->getRequestToken()][0]->getContextTag();

			/*
			 * It is important to switch to context before calling the callback
			 */
			$sys->switchContext($ctx);


			/*
			 * Call the request processing callback routine
			 */
			call_user_func($this->pool[$res->getRequestToken()][2], $res);

			/*
			 * Remove asynchronous request from pool
			 */
			$this->dequeue($this->pool[$res->getRequestToken()][0]);
		}

	}


	/**
	 * Callback for asynchronious content generation event
	 *
	 * @param evtContent processed event object 
	 * @param return void
	 */
	public function onContentReady(evtContent $e) {
		global $sys;
		/*
		 * Fetch initial request from pool
		 */
		$req = $this->pool[$e->getRequestToken()][0];

		/* 
		 * Prepare and enqueue response
		 */
		$response = new Response($req);
		$response->setData(array('headers' => $e->getResponseHeaders(), 'body' => $e->getResponseBody()));
		$this->enqueue($response);

		/*
		 * Release resources, associated with request
		 */
		$sys->releaseContext($req->getContextTag());
		unset($this->pool[$e->getRequestToken()]);
	}

	/**
	 * Callback for incoming process resource request. Always performed asynchroniously.
	 *
	 * @param Resource processed resourceobject 
	 * @return void
	 */
	public function onResourceReady(Resource $res, Request $req) {
		global $sys;
		/* 
		 * Prepare and enqueue response
		 */
		$response = new Response($req);
		$response->setData($res->serialize());
		$this->enqueue($response);

		/*
		 * Release context
		 */
		unset($this->pool[$req->getRequestToken()]);
	}



	/**
	 * Adds new request for processing
	 *
	 * This function is called directy from request class to start processing of a request. This method would also throw an exception if processing
	 * of request resulted into the remote exception on the remote node.
	 *
	 * @see Request::execSync(), Request::execAsync()
	 * @param Request Request to process
	 * @param mixed PHP callback to call when response arrives. Makes sense only for asynchronous requests.
	 * @return mixed Response to the request
	 */
	public function processOutgoingRequest(Request $req, $callback = null) {
		global $sys;

		switch ($req->getRequestType()) {
			case RequestType::RESOURCE:
				$req->setContextTag($sys->getContextTag());
			break;
		}


		if ($req->getExecType() == SYNC) {
			trigger_error('+New syncronious request execution. Type: ['.$req->getRequestType().'] Token: ['.$req->getRequestToken().']');
			$this->enqueue($req);
			trigger_error('Entering multiplexor. Expecting ['.$req->getRequestToken().']');
			try {
				/* VERY IMPORTANT !!! */

				if ($sys->context) {
					$ctx = $sys->getContextTag();
				}

				$ret = $this->multiplexor($req->getRequestToken());

				/* VERY IMPORTANT !!! */
				if (isset($ctx)) $sys->switchContext($ctx);

				trigger_error('Returned from multiplexor with ['.$req->getRequestToken().']');
				$this->dequeue($req);
				trigger_error('-Removed request from request pool');
				return $ret->getData();
			} catch (Exception $e) {
				$this->dequeue($req);
				throw $e;
			}
		}

		if ($req->getExecType() == ASYNC) {
			trigger_error('+New asyncronious request execution. Type: ['.$req->getRequestType().'] Token: ['.$req->getRequestToken().']');
			$this->enqueue($req, $callback);
			trigger_error('-Request was added to the system requests pool');
		}
		
	}

}

